package com.sinaapp.gavinzhang.GSpider.process;

import com.caucho.hessian.client.HessianRuntimeException;
import com.caucho.hessian.server.HessianServlet;
import com.sinaapp.gavinzhang.GSpider.common.IpUtil;
import com.sinaapp.gavinzhang.GSpider.common.RegisterUtil;
import com.sinaapp.gavinzhang.GSpider.common.ServletUtil;
import com.sinaapp.gavinzhang.GSpider.config.PageProcessorConfig;
import com.sinaapp.gavinzhang.GSpider.http.annotation.ServletClass;
import com.sinaapp.gavinzhang.GSpider.config.ConfigDispatcher;
import com.sinaapp.gavinzhang.GSpider.management.Register;
import com.sinaapp.gavinzhang.GSpider.management.RegistrationCenter;
import com.sinaapp.gavinzhang.GSpider.management.Stub;
import com.sinaapp.gavinzhang.GSpider.webaddress.URLDispatcher;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author gavin
 * @version 1.0
 * @date 2016/4/4
 * @description 抓取集群管理器
 */
@ServletClass(requestName = "/" + RegistrationCenter.PROCESSORDISPATCHER)
public class ProcessorManagement extends HessianServlet implements ProcessorDispatcher{
    private static final Logger logger = LoggerFactory.getLogger(ProcessorManagement.class);
    public final static int DEFAULT_PORT = 20002;
    public final static int DEFAULTEXECUTORNUM = 4;

    private static String centerIp;
    private static int centerPort;
    private static Stub processStub = Stub.newStub();

    private PageProcessorConfig pageProcessorConfig = PageProcessorConfig.newPageProcessorConfig();
    /**
     * 任务线程池
     */
    private ExecutorService taskExecutor;

    /**
     * 任务线程池数量
     */
    public int taskExecutorNum;

    private List<Process> processes;

    public ProcessorManagement() {
        this(DEFAULTEXECUTORNUM);
    }

    public ProcessorManagement(int taskExecutorNum) {
        this.taskExecutorNum = taskExecutorNum;
        taskExecutor = Executors.newFixedThreadPool(this.taskExecutorNum);
        processes = new ArrayList<>();
        for (int i = 0; i < taskExecutorNum; i++) {
            processes.add(new Process());
        }
        for (int i = 0; i < taskExecutorNum; i++) {
            processes.get(i).setPageProcessorConfig(pageProcessorConfig);
            taskExecutor.execute(processes.get(i));
        }
        try {
            flushConfig(processes);
        }catch (Exception e){
            logger.error("刷新网址/配置失败");
        }
    }

    @Override
    public void setPageProcessorConfig(PageProcessorConfig pageProcessorConfig) {
        try {
            flushConfig(processes);
        }catch (Exception e){
            logger.error("刷新网址/配置失败");
        }
    }

    private void flushConfig(List<Process> processes) throws Exception {
        Register center = RegisterUtil.getRegister(centerIp, centerPort, RegistrationCenter.REGISTRATIONCENTER);
        //配置管理器
        Stub stub = center.getConfigDispatcher();
        ConfigDispatcher configDispatcher;
        PageProcessorConfig config;
        if (stub.equals(Stub.getNoneStub())) {
            logger.warn("找不到配置管理器");
            throw new Exception();
        } else {
            configDispatcher = (ConfigDispatcher) RegisterUtil.getHessianServer(ConfigDispatcher.class, stub.getName());
            config = configDispatcher.getPageProcessorConfig();
        }

        //网址管理器
        stub = center.getURLDispatcher();
        URLDispatcher urlDispatcher = null;
        if (stub.equals(Stub.getNoneStub())) {
            logger.warn("找不到网址管理器");
            throw new Exception();
        } else {
            urlDispatcher = (URLDispatcher) RegisterUtil.getHessianServer(URLDispatcher.class, stub.getName());
        }
        config.setUrlDispatcher(urlDispatcher);
        for (Process process : processes) {
            process.setPageProcessorConfig(config);
        }
    }

    @Override
    public PageProcessorConfig getPageProcessorConfig() {
        return this.pageProcessorConfig;
    }

    @Override
    public String getStatus() {
        ThreadPoolExecutor threadPool = (ThreadPoolExecutor) this.taskExecutor;
        String status = "<br><p>";
        status +="线程池活跃线程数: "+threadPool.getActiveCount()+"<br></p>";
        return status;
    }

    @Override
    public int echo() {
        return HEARTECHO;
    }

    @Override
    public void shutdown() {
        logger.info("服务即将关闭...");
        System.exit(HEARTECHO);
    }

    public static void setUp() throws Exception {
        logger.warn("注册中心IP为:(默认)" + IpUtil.getRealIp());
        setUp(IpUtil.getRealIp());
    }

    public static void setUp(String ip) throws Exception {
        logger.warn("注册中心端口号为:(默认)" + RegistrationCenter.DEFAULT_PORT);
        setUp(ip, RegistrationCenter.DEFAULT_PORT);
    }

    public static void setUp(int localPort)throws Exception{
        logger.warn("注册中心IP为:(默认)" + IpUtil.getRealIp());
        logger.warn("注册中心端口号为:(默认)" + RegistrationCenter.DEFAULT_PORT);
        setUp(IpUtil.getRealIp(),RegistrationCenter.DEFAULT_PORT,localPort);
    }


    public static void setUp(String ip, int port) throws Exception {
        logger.warn("抓取集群管理器端口号为:(默认)" + ProcessorManagement.DEFAULT_PORT);
        setUp(ip, port, ProcessorManagement.DEFAULT_PORT);
    }

    public static void setUp(String ip, int port, int localPort) throws Exception {
        logger.info("注册中心IP为:"+ip);
        logger.info("注册中心端口号为:"+port);
        centerIp = ip;
        centerPort = port;
        //得到注册中心
        logger.info("获取注册中心服务");
        Register center = RegisterUtil.getRegister(centerIp, centerPort, RegistrationCenter.REGISTRATIONCENTER);

        //注册服务
        logger.info("注册抓取集群管理器");
        String name = "http://" + IpUtil.getRealIp() + ":" + localPort + "/" + RegistrationCenter.PROCESSORDISPATCHER+"."+processStub.getToken();
        processStub.setName(name);
        logger.info("注册中..."+processStub);
        try {
            if (center.Registrate(processStub).equals(Stub.getNoneStub())) {
                logger.error("注册失败!");
                System.exit(-1);
            }
        }catch (HessianRuntimeException e){
            logger.error("注册中心连接超时!"+centerIp+":"+centerPort);
            System.exit(-1);
        }
        logger.info("注册成功!");

        //启动jetty本地服务
        logger.info("抓取集群管理器端口号为:"+localPort);
        Server server = new Server(localPort);
        ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
        server.setHandler(context);
        logger.info("GServer 启动中...");
        ServletUtil.addServlet(context,ProcessorManagement.class,"/"+RegistrationCenter.PROCESSORDISPATCHER+"."+processStub.getToken());
        logger.info("GServer 添加RPC抓取集群管理服务成功!");
        server.start();
        logger.info("GServer 地址: " + server.getURI().toString());
        logger.info("GServer 启动成功!" + server.getState());
    }

    public static void main(String[] args) {
        try {
            ProcessorManagement.setUp("127.0.0.1");
            System.out.println("ok");
        } catch (Exception e) {
            if (e.getMessage().contains("java.net.ConnectException")) {
                logger.error("无法连接注册中心");
            }
        }

    }
}
